package flink_p1

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object FlinkTest_13_State_ValueState {


  /**
   * 检查车辆是否急加速
   * 思路：通过ValueState来记录上一次的车速度，然后与本次的对比，再记录下本次的速度
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    socketStream.map(data => {
      val arr: Array[String] = data.split(" ")
      (arr(0), arr(1).toLong)
    }).keyBy(0)
      .map(new RichMapFunction[(String, Long), (String, Long)] {


        private var lastSpeedState: ValueState[Long] = _

        override def open(configuration: Configuration): Unit = {

          val desc = new ValueStateDescriptor[Long]("car_speed", createTypeInformation[Long])
          lastSpeedState = getRuntimeContext.getState(desc)

        }

        override def map(in: (String, Long)): (String, Long) = {

          val lastSpeedValue: Long = lastSpeedState.value()
          lastSpeedState.update(in._2)
          if (lastSpeedValue != 0 && in._2 - lastSpeedValue > 30) {
            println(s"over speed: $lastSpeedValue -> ${in._2}")
          }
          in

        }

      }).print()


    env.execute()

  }


}
